package com.zn.socketReceiver;

import com.alibaba.fastjson.JSONObject;
import com.zn.asycn.CalculateKLineExecutor;
import com.zn.entity.KLineTickDto;
import com.zn.entity.OptionMarketDataField;
import com.zn.util.*;
import com.zn.vo.OptionSymbol;
import com.zn.web.dto.KLineDto;
import com.zn.web.rpc.MdRemoteCall;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by yinjiawei on 2018/7/12.
 */
public class ReaderTask extends Thread{
    private static final Logger LOGGER = LoggerFactory.getLogger(ReaderTask.class);
    private SocketStatusListener socketStatusListener;
    private static final Integer FIDLE_SIZE=30;
    private RedisTemplate<String,String> redisTemplate;
    private MongoTemplate mongoTemplate;
    private MdRemoteCall mdRemoteCall;
    private ExecutorService pool = Executors.newCachedThreadPool();
    private static final List<String> EX_DATE_NUM = Arrays.asList("A", "B", "C","D");
    private Socket socket;
    private CalculateKLineExecutor calculateKLineExecutor;

    private boolean listening;

    public ReaderTask(Socket socket,RedisTemplate<String,String> redisTemplate,
                      MongoTemplate mongoTemplate,MdRemoteCall mdRemoteCall,CalculateKLineExecutor calculateKLineExecutor) throws IOException{
        this.socket = socket;
        this.redisTemplate=redisTemplate;
        this.mongoTemplate=mongoTemplate;
        this.mdRemoteCall=mdRemoteCall;
        this.calculateKLineExecutor=calculateKLineExecutor;
    }

    /**
     * finish:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     * @throws IOException
     *
     */
    public void finish() throws IOException
    {
        listening = false;
        interrupt();
        if(socket!=null)
        {
            if(socket.isInputShutdown())
            {
                socket.shutdownInput();
            }
        }
    }

    /* (non-Javadoc)
     * @see java.lang.Runnable#run()
     */
    @Override
    public synchronized  void run()
    {
        while (listening)
        {
            try {
                this.process();
            } catch (Exception e) {
                listening = false;
                if(socketStatusListener!=null)
                {
                    int status = parseSocketStatus(e);
                    socketStatusListener.onSocketStatusChanged(socket, status, e);
                }
                e.printStackTrace();
                return;//终止线程继续运行,这里也可以使用continue
            }

        }
    }

    public void process() throws Exception {
        // Prepare date format
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd HH:mm:ss.SSS");
        dateFormat.setTimeZone(Constant.DEFAULT_TIMEZONE);
        // Prepare read buffer
        ByteBuffer bufReceive = ByteBuffer.allocate(1024 * 200);
        byte bufRead[] = new byte[1024 * 100];
        InputStream is = socket.getInputStream();
        try {
            int count = 0;
            while((count = is.read(bufRead)) > 0) {
                if(count > bufReceive.remaining()) {
                    throw new Exception("Proxy data receiver buffer overflow");
                }
                bufReceive.put(bufRead, 0, count);
                this.parseData(bufReceive);
            }
        }
        finally {
            try {is.close();}catch(Exception e) {e.printStackTrace();}
            try {socket.close();}catch(Exception e) {e.printStackTrace();}
        }
    }

    /**
     * 解析行情数据
     * @param buf 数据缓冲
     */
    protected void parseData(ByteBuffer buf) {
        buf.flip();
        byte data[] = new byte[buf.remaining()];
        buf.get(data);
        int startIndex = 0;
        for(int i = 0; i < data.length; i++) {
            if(data[i] == '\n') {
                if(i > startIndex) {
                    String item = null;
                    try {
                        item = new String(data, startIndex, i - startIndex, "GB2312");
                        if(item.startsWith("heartbeat")) {//接受心跳数据

                        }else if(item.startsWith("optionSymbol")) {
                            //接受主力合约列表的数据
							LOGGER.warn("optionSymbol" + item);
//                            this.parseOptionSymbol(item);
                        }else if(item.startsWith("tick")){
                            //接受深度行情数据
//							LOGGER.info("tick" + item);
                            this.parseDataItem(item);
                        }
                    }catch(Exception e) {
                        e.printStackTrace();
                        LOGGER.error(this.getClass() + "Proxy data receiver parse error: " + e.getMessage());
                        LOGGER.error(this.getClass() + "" + item);
                    }
                }
                startIndex = i + 1;
            }
        }
        buf.clear();
        if(startIndex < data.length) {
            buf.put(data, startIndex, data.length - startIndex);
        }
    }

    /**
     * 校验optionSymbol的key是否存在
     * @param date
     * @return
     */
    private String checkOptionKey(String date,int version){
        try {
            String nowKey=Constant.OPTION_SYMBOL_INFO + date;
            Set<Object> hashKey=redisTemplate.opsForHash().keys(nowKey);
            if(hashKey.size()==0 && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(date),1));
                version--;
                nowKey=this.checkOptionKey(yesterday,version);
            }
            return nowKey;
        }catch (Exception e){
            LOGGER.error("checkOptionKey--------->"+e);
        }
        return null;
    }



    private String [] subStringSinaData(String context){
        String [] mdInfoArray = context.split("=");
        if(mdInfoArray.length<2){
            return null;
        }
        String mdInfoElement=mdInfoArray[mdInfoArray.length-1];
        String mdInfoStr=mdInfoElement.substring(1, mdInfoElement.length() - 2);
        String [] mdInfos=mdInfoStr.split(",");
        return mdInfos;
    }
    /**
     * 处理深度行情
     * @param fromCLanguageMdInfo
     * @throws Exception
     */
    public void parseDataItem(String fromCLanguageMdInfo) throws Exception {
        //tick|最新价|持仓量|申卖价五|申卖量五|申卖价四|申卖量四|申卖价三|申卖量三|申卖价二|申卖量二|申卖价一|申卖量一|申买价一|申买量一|申买价二|申买量二|申买价三|申买量三|申买价四|申买量四|申买价五|申买量五|行情时间|最高价|最低价|成交量|成交额|昨收盘|今开盘|涨停价|跌停价
        // LOGGER.info("接受到C++的行情数据：" + fromCLanguageMdInfo);
        /**根据管道符分割**/
        String [] mdDetailArray=fromCLanguageMdInfo.split("\\|");
        String instrumentCode=mdDetailArray[1],lastPrice=mdDetailArray[2],openInterest=mdDetailArray[3],
                askPrice5=mdDetailArray[4],askVolume5=mdDetailArray[5],askPrice4=mdDetailArray[6],askVolume4=mdDetailArray[7],
                askPrice3=mdDetailArray[8],askVolume3=mdDetailArray[9],askPrice2=mdDetailArray[10],askVolume2=mdDetailArray[11],
                askPrice1=mdDetailArray[12],askVolume1=mdDetailArray[13],bidPrice1=mdDetailArray[14],bidVolume1=mdDetailArray[15],
                bidPrice2=mdDetailArray[16],bidVolume2=mdDetailArray[17],bidPrice3=mdDetailArray[18],bidVolume3=mdDetailArray[19],
                bidPrice4=mdDetailArray[20],bidVolume4=mdDetailArray[21],bidPrice5=mdDetailArray[22],bidVolume5=mdDetailArray[23],
                updateTime=mdDetailArray[24],highPrice=mdDetailArray[25],lowPrice=mdDetailArray[26],volume=mdDetailArray[27],
                turnover=mdDetailArray[28],preClosePrice=mdDetailArray[29],openPrice=mdDetailArray[30],upperLimitPrice=mdDetailArray[31],
                LowerLimitPrice=mdDetailArray[32];

        KLineTickDto kLineTickDto=new KLineTickDto(instrumentCode, lastPrice, volume, turnover, openInterest, updateTime);

        if ("10001557".equals(instrumentCode)){
//            LOGGER.info("接受到C++的行情数据：" + fromCLanguageMdInfo);
            Date updateTime1=DateUtils.convertStringToDate("HH:mm:ss", kLineTickDto.getUpdateTime());
            String minNum=DateUtils.formatDate("HH:mm", updateTime1);
            String hashKey=instrumentCode+"-"+minNum;
            calculateKLineExecutor.calculateOneMinKLine(hashKey,kLineTickDto);
        }

    }

    public int randomNum(){
        return (int)((Math.random()*9+1)*10);
    }

    private Object beforeOptionSymbolObj(String nowDate,String instrumentCode,int version){
        try {
            String optionSymbolKey=Constant.OPTION_SYMBOL_INFO + nowDate;
            Object optionSymbolObj=redisTemplate.opsForHash().get(optionSymbolKey,instrumentCode);
            if(optionSymbolObj==null && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(nowDate), 1));
                version--;
                optionSymbolObj=this.beforeOptionSymbolObj(yesterday, instrumentCode, version);
            }
            return optionSymbolObj;
        }catch (Exception e){
            LOGGER.error("beforeOptionSymbolObj--------->"+e);
        }
        return null;
    }

    private Object beforeTradeday(String targetCode,String date,String instrumentMonth,String instrumentCode,int version){
        try {
            String tickKey=Constant.OPTION_TICKL_INFO+targetCode+":"+instrumentMonth+":"+date;
            Object tickObject=redisTemplate.opsForHash().get(tickKey,instrumentCode);
            if(tickObject==null && version>=0){
                String yesterday=DateUtils.convertDateToString(DateUtils.minusDay(DateUtils.convertStringToDate(date), 1));
                version--;
                tickObject=this.beforeTradeday(targetCode,yesterday,instrumentMonth,instrumentCode,version);
            }
            return tickObject;
        }catch (Exception e){
            LOGGER.error("beforeTradeday--------->",e);
        }
        return null;
    }

    /**
     * 采用线程池去刷新redis最新价
     * @param nowDate
     * @param optionMarketDataField
     */
    public void sendRedisLastPrice(String targetCode,String nowDate,OptionMarketDataField optionMarketDataField) {
        pool.execute(new Thread(() -> {
            try {
                LOGGER.info("sendRedisLastPrice:" + optionMarketDataField.getInstrumentCode() + "," + optionMarketDataField.toString());
                /**主力合约json**/
                Object mainContractToJson= JSONObject.toJSON(optionMarketDataField);
                /**推送完成之后更新首页主力合约列表的合约最新值**/
                redisTemplate.opsForHash().put(Constant.OPTION_TICKL_INFO+targetCode+":"+optionMarketDataField.getInstrumentMonth()+":"+nowDate,optionMarketDataField.getInstrumentCode(),mainContractToJson.toString());
                redisTemplate.expire(Constant.OPTION_TICKL_INFO+targetCode+":"+optionMarketDataField.getInstrumentMonth()+":"+nowDate, 10, TimeUnit.DAYS);
            }catch (Exception e){
                e.printStackTrace();
                LOGGER.warn("sendRedisLastPrice error ..........." + e);
            }
        }));
    }
    private int parseSocketStatus(Exception e)
    {
        if(SocketException.class.isInstance(e))
        {
            String msg = e.getLocalizedMessage().trim();
            if("Connection reset".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_RESET;
            }
            else if("Socket is closed".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_CLOSE;
            }
            else if("Broken pipe".equalsIgnoreCase(msg))
            {
                return SocketStatusListener.STATUS_PIP_BROKEN;
            }

        }
        return SocketStatusListener.STATUS_UNKOWN;
    }

    /**
     * listen:(这里用一句话描述这个方法的作用). <br/>
     * TODO(这里描述这个方法适用条件 – 可选).<br/>
     *
     */
    public void startListener(SocketStatusListener ssl) {
        listening = true;
        this.socketStatusListener = ssl;
        start();
    }
}
